package com.vf.cloud.paas.app.agent.common.server.signalling.handler;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import com.alibaba.fastjson.JSONObject;
import com.jfinal.kit.JsonKit;
import com.jfinal.kit.StrKit;
import com.vf.cloud.paas.app.agent.common.server.signalling.pool.PlayerPool;
import com.vf.cloud.paas.app.agent.common.server.signalling.pool.StreamerPool;
import com.vf.cloud.paas.app.agent.common.server.signalling.util.SignallingUtil;
import com.vf.cloud.paas.app.agent.common.server.signalling.util.UETimerUtil;
import com.vf.cloud.paas.common.cirrus.Cirrus;
import com.vf.cloud.paas.common.service.IRedis;
import com.vf.cloud.paas.common.util.PrefixUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpMessage;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class StreamerChannelInboundHandlerHandler extends SimpleChannelInboundHandler<Object> {

	
	private IRedis redisImpl;
	private String awsIp;
	private boolean enabled;
	private int streamerPort;
	private int stunPort;
	private int turnPort;
	private String username;
	private String password;
	// private String localIp;
	private int idleTimeout;

	public StreamerChannelInboundHandlerHandler(IRedis redisImpl,int streamerPort,String awsIp, boolean enabled, int stunPort, int turnPort,
			String username, String password, String localIp, int idleTimeout) {
		super();
		this.awsIp = awsIp;
		this.enabled = enabled;
		this.stunPort = stunPort;
		this.turnPort = turnPort;
		this.username = username;
		this.password = password;
		// this.localIp = localIp;
		this.idleTimeout = idleTimeout;
		this.redisImpl=redisImpl;
		this.streamerPort=streamerPort;
		
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		log.info(String.format("Streamer connected:" + ctx.channel().remoteAddress()));
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		log.error("Streamer disconnected:" + ctx.channel().remoteAddress());
		SignallingUtil.onStreamerDisconnected(ctx.channel().id().asLongText(),redisImpl);
	}

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		ctx.flush();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		log.error("Streamer connection error:", cause);
		ctx.close();
		// SignallingUtil.onStreamerDisconnected(ctx.channel().id().asLongText());
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
		if (msg instanceof FullHttpRequest) {
			handleHttpRequest(ctx, (FullHttpRequest) msg);
		} else {
			handlerWebSocketFrame(ctx, msg);
		}
	}

	protected WebSocketServerHandshaker handshaker;

	/**
	 * 处理 FullHttpRequest
	 * 
	 * @param ctx
	 * @param request
	 */
	protected void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
		if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
			sendHttpResponse(ctx, req,
					new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
			return;
		}

		String uri = req.uri();
		WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
				"ws://" + req.headers().get(HttpHeaderNames.HOST) + uri, null, false);
		handshaker = wsFactory.newHandshaker(req);
		if (handshaker == null) {
			WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
		} else {
			handshaker.handshake(ctx.channel(), req);
		}

		uri = uri.replaceFirst("\\/", "");
		String[] params = uri.split("\\/", -1);
		if (params == null || params.length < 1) {
			log.info("appid>> is null");
			ctx.close();
			StreamerPool.closeByChannelHandlerContext(ctx);
			return;
		}
		
		String token = params[0];
		if(StrKit.isBlank(token)) {
			ctx.close();
			log.info("appid>> is null");
			StreamerPool.closeByChannelHandlerContext(ctx);
			return;
		}
		
		if(!redisImpl.hasKey(token)) {
			ctx.close();
			log.info("已超时渲染，主动关闭.");
			StreamerPool.closeByChannelHandlerContext(ctx);
			return;
		}
		
		String GPUIndex=redisImpl.get(token);
		
		StreamerPool.put(token, ctx);
		sendConfig(ctx);
		UETimerUtil.createTask(token, this.idleTimeout);

		if (PlayerPool.containsToken(token)) {
			JSONObject json = new JSONObject();
			json.put("type", "playerConnected");
			json.put("playerId", token);
			json.put("dataChannel", true);
			json.put("sfu", false);
			send(ctx, JsonKit.toJson(json));
		}
		Cirrus.renderClient.sendGpuInfo(GPUIndex.replaceFirst(PrefixUtil.GPU_Free, PrefixUtil.GPU_Used),GPUIndex);
	}
	


	public static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
		if (res.status().code() != 200) {
			ByteBuf buf = Unpooled.copiedBuffer(res.status().code() + "", CharsetUtil.UTF_8);
			res.content().writeBytes(buf);
			buf.release();
		}
		// 如果是非Keep-Alive，关闭连接
		if (!HttpUtil.isKeepAlive((HttpMessage) req.protocolVersion()) || res.status().code() != 200) {
			ctx.channel().writeAndFlush(res).addListener(ChannelFutureListener.CLOSE);
		}
	}

	private void handlerWebSocketFrame(ChannelHandlerContext ctx, Object frame) {
		// 判断是否关闭链路的指令
		if (frame instanceof CloseWebSocketFrame) {
			CloseWebSocketFrame closeWebSocketFrame = (CloseWebSocketFrame) frame;
			handshaker.close(ctx.channel(), (CloseWebSocketFrame) closeWebSocketFrame.retain());
			return;
		}
		// 判断是否ping消息
		if (frame instanceof PingWebSocketFrame) {
			return;
		}

		if (frame instanceof TextWebSocketFrame) {

			TextWebSocketFrame textWebSocketFrame = (TextWebSocketFrame) frame;
			JSONObject rawMsg = null;
			try {
				rawMsg = JSONObject.parseObject(textWebSocketFrame.text());
			} catch (Exception e) {
				log.error(String.format("[%s]>> cannot parse Streamer message:%s", 1008, e));
				ctx.close();
			}
			if (rawMsg == null) {
				return;
			}
			String type = rawMsg.getString("type");
			if (StrKit.equals(type, "ping")) {
				SignallingUtil.sendPong(ctx, rawMsg.getLong("time"));
				return;
			}

			String playerId = rawMsg.get("playerId").toString();
			rawMsg.remove(playerId);
			if (StrKit.equals(type, "offer")) {
				SignallingUtil.sendMessageToPlayer(playerId, rawMsg);
			} else if (StrKit.equals(type, "answer")) {
				SignallingUtil.sendMessageToPlayer(playerId, rawMsg);
			} else if (StrKit.equals(type, "iceCandidate")) {
				SignallingUtil.sendMessageToPlayer(playerId, rawMsg);
			} else if (StrKit.equals(type, "disconnectPlayer")) {
				SignallingUtil.disconnectedPlayer(playerId);
			} else {
				log.error("unsupported Streamer(%s) message type:%s",streamerPort, type);
			}
		}
	}

	/**
	 * 向客户端发送心跳
	 * 
	 * @param ctx
	 * @param evt
	 * @throws Exception
	 */
	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		if (evt instanceof IdleStateEvent) {
			IdleStateEvent event = (IdleStateEvent) evt;
			if (event.state().equals(IdleState.READER_IDLE)) {
				SignallingUtil.sendPing(ctx, System.currentTimeMillis());
			} else if (event.state().equals(IdleState.WRITER_IDLE)) {
				SignallingUtil.sendPing(ctx, System.currentTimeMillis());
			} else if (event.state().equals(IdleState.ALL_IDLE)) {
				SignallingUtil.sendPing(ctx, System.currentTimeMillis());
			}
		}
	}

	public void sendConfig(ChannelHandlerContext ctx) {
		Map<String, Object> clientConfig = new HashMap<String, Object>();
		clientConfig.put("type", "config");

		if (enabled) {
			Map<String, Object> peerConnectionOptions = new HashMap<String, Object>();

			List<Object> iceServers = new LinkedList<Object>();
			Map<String, Object> iceServer = new HashMap<String, Object>();

			List<Object> urls = new LinkedList<Object>();
			urls.add(String.format("stun:%s:%s", awsIp, stunPort));
			urls.add(String.format("turn:%s:%s", awsIp, turnPort));

			iceServer.put("urls", urls);
			iceServer.put("username", username);
			iceServer.put("credential", password);
			iceServers.add(iceServer);

			peerConnectionOptions.put("iceServers", iceServers);
			peerConnectionOptions.put("sdpSemantics", "unified-plan");
			peerConnectionOptions.put("offerExtmapAllowMixed", false);
			clientConfig.put("peerConnectionOptions", peerConnectionOptions);
		} else {
			clientConfig.put("peerConnectionOptions", new HashMap<String, Object>());
		}

		send(ctx, JsonKit.toJson(clientConfig));
	}

	public void send(ChannelHandlerContext ctx, String json) {
		if (ctx != null) {
			ctx.channel().writeAndFlush(new TextWebSocketFrame(json)).addListener(new ChannelFutureListener() {
				@Override
				public void operationComplete(ChannelFuture future) throws Exception {
					if (future.isSuccess()) {
						// log.info(String.format("-> Streamer:%s", json));
					} else {
						// log.info(String.format("-> Streamer isError :%s", json));
					}
				}
			});
		}
	}

}
